package top.whysu.edevice.emqx;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import org.eclipse.paho.client.mqttv3.MqttAsyncClient;
import org.eclipse.paho.client.mqttv3.MqttConnectOptions;
import org.eclipse.paho.client.mqttv3.MqttException;
import org.eclipse.paho.client.mqttv3.MqttMessage;
import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import top.whysu.edevice.constant.CodeConstant;
import top.whysu.edevice.constant.EmqConstant;
import top.whysu.edevice.constant.SystemConfigKey;
import top.whysu.edevice.dto.emqx.MqttBaseEntity;
import top.whysu.edevice.dto.emqx.MqttBaseRequest;
import top.whysu.edevice.global.ThreadPoolBuilder;
import top.whysu.edevice.service.system.SystemConfigService;
import top.whysu.edevice.utils.Coder;
import top.whysu.edevice.utils.SpringUtil;

/**
 * 与设备交互的EMQ客户端
 */
public class DeviceMqttClient {

    private static final Logger LOGGER = LoggerFactory.getLogger(DeviceMqttClient.class);
    private SystemConfigService systemConfigService = SpringUtil.getBean(SystemConfigService.class);
    private ThreadPoolBuilder threadPoolBuilder = SpringUtil.getBean(ThreadPoolBuilder.class);

    private static DeviceMqttClient deviceMqttClient;
//    private MqttClient client;
//    private MqttConnectOptions options;

    private MqttAsyncClient client;
    private MqttConnectOptions options;

    public MqttAsyncClient getClient() {
        return client;
    }

    /**
     * 单例
     *
     * @return DeviceMqttClient
     */
    public synchronized static DeviceMqttClient getInstance() {
        if (null == deviceMqttClient) {
            deviceMqttClient = new DeviceMqttClient();
        }
        return deviceMqttClient;
    }

    /**
     * 连接EMQ服务器
     */
    public void connect() {
        if (isMqttConnected()) {
            LOGGER.info("执行设备EMQ的connect，检测到已连接");
            return;
        }
        LOGGER.info("执行设备EMQ的connect");
        try {
            // 连接URL
            String host = EmqConstant.Config.NO_SECURITY_NET_PREFIX + EmqConstant.Config.LOCAL_IP + ":" + EmqConstant.Config.NO_SECURITY_PORT;
//            client = new MqttClient(host, Constant.Emqtt.Config.CLIENT_ID, new MemoryPersistence());
            client = new MqttAsyncClient(host, EmqConstant.Config.CLIENT_ID, new MemoryPersistence());
            options = new MqttConnectOptions();
            options.setMaxInflight(5000);
            // 设置是否清空session,这里如果设置为false表示服务器会保留客户端的连接记录，设置为true表示每次连接到服务器都以新的身份连接
            options.setCleanSession(true);
            // 设置连接超时时间，单位为秒
            options.setConnectionTimeout(30);
            // 设置会话心跳时间 单位为秒 服务器会每隔120秒的时间向客户端发送个消息判断客户端是否在线，但这个方法并没有重连的机制
            options.setKeepAliveInterval(120);
            // 设置自动重连
            options.setAutomaticReconnect(false);
            options.setUserName(EmqConstant.Config.USERNAME);
            options.setPassword(EmqConstant.Config.PASSWORD.toCharArray());
            // 设置接收回调
            client.setCallback(new DeviceSubcribeCallBack());
            client.connect(options);
        } catch (Exception e) {
            LOGGER.error("连接EMQ服务器异常", e);
            try {
                client.close();
            } catch (MqttException mqttException) {
            }
        }
    }

    /**
     * 尝试建立连接
     */
    public synchronized void reconnect() {
        try {
            if (!isMqttConnected()) {
                if (null != client) {
                    LOGGER.info("reconnect");
                    disConnect();
                    connect();
                } else {
                    connect();
                }
            }
        } catch (Exception e) {
            LOGGER.error("尝试建立连接异常", e);
        }
    }

    /**
     * 发布明文信息给设备
     *
     * @param mqttBaseEntity 请求数据
     * @param deviceId       设备DID
     */
    public void publishNoEncrpyMsg(MqttBaseEntity mqttBaseEntity, String deviceId, long sessionId) {
        Thread t = new MqttPublishChannel(1, mqttBaseEntity, deviceId, false, sessionId);
        threadPoolBuilder.getThreadPool(ThreadPoolBuilder.THREADPOOL_LOCAL_KEY_PUBLISHMSG_TO_DEVICE).execute(t);
    }

    /**
     * 发布密文信息给设备
     *
     * @param mqttBaseEntity 请求数据（需设置参数和方法）
     * @param deviceId       设备DID
     * @param sessionId      sessionId
     */
    public void publishEncrpyMsg(MqttBaseEntity mqttBaseEntity, String deviceId, long sessionId) {
        Thread t = new MqttPublishChannel(1, mqttBaseEntity, deviceId, true, sessionId);
        threadPoolBuilder.getThreadPool(ThreadPoolBuilder.THREADPOOL_LOCAL_KEY_PUBLISHMSG_TO_DEVICE).execute(t);
    }

    /**
     * 消息发送
     *
     * @param message 消息
     * @throws Exception 异常
     */
    private synchronized void publish(MqttMessage message, String topic) throws Exception {
        if (!isMqttConnected()) {
            LOGGER.info("消息发送时，未连接EMQTT");
            return;
        }
        client.publish(topic, message);
    }

    /**
     * 多线程消息发送类
     */
    class MqttPublishChannel extends Thread {
        int qos;
        MqttBaseEntity mqttBaseEntity;
        String deviceId;
        boolean isEncrypt;
        long sessionId;

        @Override
        public void run() {
            super.run();
            MqttMessage mqttMessage = new MqttMessage();
            String topic;
            try {
                mqttBaseEntity.setVersion(EmqConstant.Config.VERSION);
                mqttBaseEntity.setSrc(EmqConstant.Config.CLIENT_ID);
                mqttBaseEntity.setDst(deviceId);
                mqttBaseEntity.setSeq(sessionId);
                mqttMessage.setQos(qos);
                LOGGER.info("发布消息给设备：{}，参数为：{}", JSONObject.toJSONString(mqttMessage), JSON.toJSONString(mqttBaseEntity));
                // 是否加密判断
                if (isEncrypt) {
                    topic = EmqConstant.Topic.DEVICE_PREFIX + deviceId;
                    Coder coder = new Coder(EmqConstant.Config.GLOBAL_VERSION);
                    mqttMessage.setPayload(coder.encrypt(JSON.toJSONString(mqttBaseEntity).getBytes("UTF-8")));
                } else {
                    topic = EmqConstant.Topic.DEVICE_PREFIX + deviceId + "/" + EmqConstant.Topic.PLAIN;
                    mqttMessage.setPayload(JSON.toJSONString(mqttBaseEntity).getBytes("UTF-8"));
                }
                LOGGER.debug("publish执行前{}", this.getName());
                publish(mqttMessage, topic);
                Thread.sleep(10);
            } catch (Exception e) {
                LOGGER.error("消息发送异常", e);
            }
            LOGGER.debug("当前线程执行完成{}", this.getName());
        }

        private MqttPublishChannel(int qos, MqttBaseEntity mqttBaseEntity, String deviceId, boolean isEncrypt, long sessionId) {
            this.qos = qos;
            this.mqttBaseEntity = mqttBaseEntity;
            this.deviceId = deviceId;
            this.isEncrypt = isEncrypt;
            this.sessionId = sessionId;
        }
    }

    /**
     * 关闭mqtt连接
     */
    private void disConnect() {
        try {
            if (!client.isConnected()) {
                // 断开连接
                client.disconnect();
            } else {
                LOGGER.info("执行设备EMQ的disconnect时，检测到当前状态为未连接");
            }
        } catch (Exception e) {
            LOGGER.error("关闭EMQ连接异常", e);
        }
    }

    /**
     * 判断mqtt是否正常连接
     *
     * @return 连接状态
     */
    private boolean isMqttConnected() {
        boolean hasConnected = false;
        if (null != client) {
            hasConnected = this.client.isConnected();
        }
        return hasConnected;
    }
}
